// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <algorithm>
#include <iterator>
#include <map>
#include <memory>
#include <optional>
#include <utility>
#include <vector>

#include <boost/range/adaptor/reversed.hpp>

#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/feed_util/book.h"

class BookBuilder;

class Transaction {
  friend class BookBuilder;
  Transaction(BookBuilder* book, bool enable_delta_book);

 public:
  void UpdateAsk(double price, double qty);
  void UpdateBid(double price, double qty);
  // bbo
  void UpdateAsk0(double price, double qty);
  void UpdateBid0(double price, double qty);
  void Clear();  // == snapshot
  void BboClear();  // == light bbo
  bool IsCrossedBook();
  void UpdateTimestamp(int64_t timestamp);

  ~Transaction();

 private:
  BookBuilder* book_;
  bool enable_delta_book_;
  std::optional<double> min_ask_price_ = {};
  std::optional<double> max_bid_price_ = {};
};

class BookBuilder : public IBook {
  friend class Transaction;

 public:
  enum BookOption {
    SNAPSHOT,
    DELTA,
    LIGHT_BBO
  };
  // if enable_diff_book=true,
  // we keep the delta book as well
  explicit BookBuilder(bool enable_diff_book) : enable_diff_book_(enable_diff_book) {}

  BookBuilder(const BookBuilder& from_bb, size_t max_book_level)
      : enable_diff_book_(from_bb.enable_diff_book_) {
    size_t nask = std::min(from_bb.asks_.size(), max_book_level);
    size_t nbid = std::min(from_bb.bids_.size(), max_book_level);

    asks_.insert(from_bb.AskBegin(), std::next(from_bb.AskBegin(), nask));
    // Q: this is reverse order, is it okay?
    bids_.insert(from_bb.BidBegin(), std::next(from_bb.BidBegin(), nbid));

    CHECK_EQ(asks_.size(), nask);
    CHECK_EQ(bids_.size(), nbid);
  }

  void CleanUp(std::optional<double> min_ask_price, std::optional<double> max_bid_price) {
    // clean up irregular points
    if (min_ask_price) {
      // remove all bids >= min_ask_price
      bids_.erase(bids_.lower_bound(*min_ask_price), bids_.end());
    }
    if (max_bid_price) {
      // remove all asks <= max_bid_price
      asks_.erase(asks_.begin(), asks_.upper_bound(*max_bid_price));
    }
  }

  /// IBook interfaces
  bool IsSnapshot() const override { return is_snapshot_; }
  int64_t Timestamp() const override { return timestamp_; }

  std::optional<double> AskQty(double price) const override {
    auto it = asks_.find(price);
    if (it == asks_.end()) {
      return {};
    } else {
      return std::optional(it->second);
    }
  }
  std::optional<double> BidQty(double price) const override {
    auto it = bids_.find(price);
    if (it == bids_.end()) {
      return {};
    } else {
      return std::optional(it->second);
    }
  }

  std::optional<PriceQty> Ask0() const override {
    if (asks_.empty()) {
      return {};
    } else {
      auto it = asks_.begin();
      return std::optional(PriceQty {it->first, it->second});
    }
  }
  std::optional<PriceQty> Bid0() const override {
    if (bids_.empty()) {
      return {};
    } else {
      auto it = bids_.rbegin();
      return std::optional(PriceQty {it->first, it->second});
    }
  }
  std::optional<PriceQty> AskN(size_t n) const override {
    if (asks_.size() <= n) {
      return {};
    } else {
      auto it = std::next(asks_.begin(), n);
      return std::optional(PriceQty {it->first, it->second});
    }
  }
  std::optional<PriceQty> BidN(size_t n) const override {
    if (bids_.size() <= n) {
      return {};
    } else {
      auto it = std::next(bids_.rbegin(), n);
      return std::optional(PriceQty {it->first, it->second});
    }
  }
  AskIterator AskBegin() const override { return asks_.begin(); }
  AskIterator AskEnd() const override { return asks_.end(); }
  BidIterator BidBegin() const override { return bids_.rbegin(); }
  BidIterator BidEnd() const override { return bids_.rend(); }

  const std::map<double, double>& Asks() const override { return asks_; }
  boost::reversed_range<const std::map<double, double>> Bids() const override {
    return boost::adaptors::reverse(bids_);
  }

  size_t AskSize() const override { return asks_.size(); }
  size_t BidSize() const override { return bids_.size(); }

  /// return ask0..askn (n+1 elements)
  std::vector<PriceQty> Asks(int n) const override {
    if (n > static_cast<int>(asks_.size())) {
      n = asks_.size();
    }
    std::vector<PriceQty> v;
    std::for_each(asks_.begin(), std::next(asks_.begin(), n), [&v](auto pair) {
      v.emplace_back(pair.first, pair.second);
    });
    return v;
  }

  /// return bid0..bidn (n+1 elements)
  std::vector<PriceQty> Bids(int n) const override {
    if (n > static_cast<int>(bids_.size())) {
      n = bids_.size();
    }
    std::vector<PriceQty> v;
    std::for_each(bids_.rbegin(), std::next(bids_.rbegin(), n), [&v](auto pair) {
      v.emplace_back(pair.first, pair.second);
    });
    return v;
  }

  const std::vector<PriceQty>& DeltaAsks() const override {
    CHECK_THROW(enable_diff_book_);
    return delta_asks_;
  }
  const std::vector<PriceQty>& DeltaBids() const override {
    CHECK_THROW(enable_diff_book_);
    return delta_bids_;
  }

  void Clear() {
    asks_.clear();
    bids_.clear();
  }

  [[deprecated("This method is considered to have more cache misses")]]
  // line break
  void
  OnSnapshot(
      int64_t timestamp,
      const std::vector<PriceQty>& asks,
      const std::vector<PriceQty>& bids) {
    Clear();
    timestamp_ = timestamp;
    for (auto ask : asks) {
      asks_.emplace(ask.price, ask.qty);
    }
    for (auto bid : bids) {
      bids_.emplace(bid.price, bid.qty);
    }
  }

  [[deprecated("This method is considered to have more cache misses")]]
  // line break
  void
  OnDiff(int64_t timestamp, const std::vector<PriceQty>& asks, const std::vector<PriceQty>& bids) {
    std::optional<double> min_ask_price = {};
    std::optional<double> max_bid_price = {};

    timestamp_ = timestamp;
    for (auto ask : asks) {
      asks_.insert(std::make_pair(ask.price, ask.qty));
      if (!min_ask_price || ask.price < *min_ask_price) {
        min_ask_price.emplace(ask.price);
      }
    }
    for (auto bid : bids) {
      bids_.insert(std::make_pair(bid.price, bid.qty));
      if (!max_bid_price || bid.price > *max_bid_price) {
        max_bid_price.emplace(bid.price);
      }
    }

    CleanUp(min_ask_price, max_bid_price);
  }
  Transaction NewTransaction(int64_t timestamp) {
    timestamp_ = timestamp;
    return Transaction(this, enable_diff_book_);
  }

  bool SampleByFrequency(
      const std::optional<PriceQty>& bid_post,
      const std::optional<PriceQty>& ask_post,
      const double sampling_frequency_bps) {
    bool sampled = false;
    if (!bid_post || !ask_post) {
      sampled = false;
    } else if (!bid_presampled_ || !ask_presampled_) {
      sampled = true;
    } else {
      double pre_midp = 0.5 * (bid_presampled_->price + ask_presampled_->price);
      double post_midp = 0.5 * (bid_post->price + ask_post->price);
      sampled = std::abs(FDiffBps(pre_midp, post_midp)) > sampling_frequency_bps;
    }
    if (sampled) {
      bid_presampled_ = *bid_post;
      ask_presampled_ = *ask_post;
    }
    return sampled;
  }

  void SetTimestamp(int64_t timestamp) {
    timestamp_ = timestamp;
  }

 private:
  static double FDiffBps(double prc1, double prc2) {
    double mid_of_2 = 0.5 * (prc1 + prc2);
    // can't be both zero. almost safe to check
    CHECK_GT(mid_of_2, 0);
    return (prc1 - prc2) / mid_of_2 * 1e4;
  }

 protected:
  bool is_snapshot_ = false;
  int64_t timestamp_ = 0;
  std::map<double, double> asks_;
  std::map<double, double> bids_;

  // delta book
  bool enable_diff_book_;
  // mainly used for fastfeed gen
  std::vector<PriceQty> delta_asks_;
  std::vector<PriceQty> delta_bids_;

  std::optional<PriceQty> bid_presampled_;
  std::optional<PriceQty> ask_presampled_;
};

inline std::ostream& operator<<(std::ostream& os, const PriceQty& pq) {
  os << "PriceQty";
  os << "(price=";
  os << pq.price;
  os << ",qty=";
  os << pq.qty;
  os << ")";
  return os;
}

template <class T>
inline std::ostream& operator<<(std::ostream& os, const std::optional<T>& o) {
  if (o) {
    os << o.value();
  } else {
    os << "None";
  }
  return os;
}

inline std::ostream& operator<<(std::ostream& os, const IBook& book) {
  os << "IBook";
  os << "(ask0=";
  os << book.Ask0();
  os << ",bid0=";
  os << book.Bid0();
  os << ")";
  return os;
}
